package com.xm.queue.task;

import com.xm.limiter.Limiter;
import com.xm.limiter.SlidingWindowLimiter02;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 任务工具类
 *  拆分：
 *    线程池
 *    任务队列
 *    工人
 *    任务管理者
 *    工具类
 *  过程：
 *    工具类->任务管理者单例->通过线程池->启用工人->生产、消费任务队列
 */
@Slf4j
public class TaskUtil {
    private TaskUtil() {
    }
    public static final int CPU_CORES;// 最大核心线程数
    public static final int CORE_POOL_SIZE;// 核心线程数
    private static final int MAXIMUM_POOL_SIZE;// 最大线程数
    private static final long KEEP_ALIVE_TIME = 60L;// 存活时间
    public static final int LIMIT = 100;// QPS

    private static final int CAPACITY = 100;// 队列最大容量
    private static final int RETRY_TIMES = 3;// 重试次数
    public static volatile AtomicInteger count = new AtomicInteger(0);
    public static volatile AtomicInteger realPutCount = new AtomicInteger(0);
    public static volatile AtomicInteger realPutDelayCount = new AtomicInteger(0);
    public static volatile AtomicInteger realPollCount = new AtomicInteger(0);
    private static final Limiter LIMITER_PUT;// 限流存放任务
    private static final Limiter LIMITER_POLL;// 限流拉取任务

    static {
        CPU_CORES = Runtime.getRuntime().availableProcessors();
        CORE_POOL_SIZE = CPU_CORES / 2;
        MAXIMUM_POOL_SIZE = CORE_POOL_SIZE + 2;
        LIMITER_PUT = new SlidingWindowLimiter02(LIMIT, 1000, 100);// 平均每10毫秒接收 1 个请求
        LIMITER_POLL = new SlidingWindowLimiter02(LIMIT * CORE_POOL_SIZE, 1000, 100);// 平均每10毫秒消费 CORE_POOL_SIZE 个请求
    }

    /**
     * 任务计数器 加1
     */
    public static int incrementAndGet() {
        return count.incrementAndGet();
    }

    /**
     * 任务计数器 累计
     */
    public static int count() {
        return count.intValue();
    }

    public static final BlockingQueue<Runnable> QUEUE = new LinkedBlockingQueue<>(CAPACITY);// 任务队列
    public static final BlockingQueue<Runnable> DEAD_QUEUE = new LinkedBlockingQueue<>(CAPACITY);// 任务死信队列
    /**
     * 工人线程池
     */
    public static final ExecutorService SERVICE = new ThreadPoolExecutor(
            CORE_POOL_SIZE,
            MAXIMUM_POOL_SIZE,
            KEEP_ALIVE_TIME,
            TimeUnit.SECONDS,
            QUEUE,
            new ThreadPoolExecutor.CallerRunsPolicy());
//    public static final ExecutorService SERVICE = Executors.newFixedThreadPool(4);

    private static final Lock lock = new ReentrantLock();
    private static final Condition putCon = lock.newCondition();
    private static final Condition pollCon = lock.newCondition();

    public static void testLimiterPut(String no) {
        boolean putTA = LIMITER_PUT.tryAcquire();
        log.info("LIMITER_PUT[{}] ={}", no, putTA);
        realPutCount.incrementAndGet();
    }

    public static void testLimiterPoll(String no) {
        boolean pollTA = LIMITER_POLL.tryAcquire();
        log.info("LIMITER_POLL[{}] ={}", no, pollTA);
        realPollCount.incrementAndGet();
    }

    /**
     * 添加任务
     * @param task 实现 Runnable 接口的任务
     * @return true 成功入列
     */
    public static boolean put(Runnable task) {
        try {
            if (LIMITER_PUT.tryAcquire()) {
                QUEUE.put(task);
                realPutCount.incrementAndGet();
//                log.info("添加任务[{}]", task);
                return true;
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return false;
    }

    /**
     * 拉取任务
     * @return 返回一个队列任务
     */
    public static Runnable poll() {
        if (LIMITER_POLL.tryAcquire()) {
            return QUEUE.poll();
        }
        return null;
    }

    /**
     * 非满
     * @return true 队列不满
     */
    public static boolean notFull() {
        return QUEUE.size() < CAPACITY;
    }

    /**
     * 非空
     * @return true 队列有值
     */
    public static boolean notEmpty() {
        return !QUEUE.isEmpty();
    }

    /**
     * 添加任务 直至成功
     * @param task 实现 Runnable 接口的任务
     */
    public static void putNew(Runnable task) {
        lock.lock();
        try {
            while (!notFull()) {// 当队列已满
                if (LIMITER_PUT.tryAcquire()) {// 且获取到资格
//                    log.warn("实际延迟接收任务：{}", task);// 由于消费者一直消费，所以会导致测试realPutDelayCount值比预计结果要小
                    realPutDelayCount.incrementAndGet();
                    putCon.await();// 需配合 lock 使用
//                    condition.wait();// 需配合 synchronized 使用
                }
            }
            while (true) {
                if (LIMITER_PUT.tryAcquire()) {
                    QUEUE.put(task);
                    realPutCount.incrementAndGet();
                    if (notEmpty()) {
                        pollCon.signalAll();
                    }
                    break;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 拉取任务 直至成功
     * @return 返回一个队列任务
     */
    public static Runnable pollNew() {
        lock.lock();
        try {
            while (!notEmpty()) {
                if (LIMITER_POLL.tryAcquire()) {
                    pollCon.await();
                }
            }

            while (true) {
                if (LIMITER_POLL.tryAcquire()) {
                    Runnable task = QUEUE.poll();
                    if (notFull()) {
                        putCon.signalAll();
                    }
                    return task;
                }
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 添加任务，循环 put 直至成功入列
     * @param task 实现 Runnable 接口的任务
     */
    public static void putAlways(Runnable task) {
        try {
            while (true) {
                if (put(task)) {
                    realPutDelayCount.incrementAndGet();
                    log.warn("添加任务，循环 put 直至成功入列，计数 {}", realPutDelayCount.intValue());
                    break;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 添加任务，循环 offer 直至成功入列
     * @param task 实现 Runnable 接口的任务
     */
    public static void offerAlways(Runnable task) {
        try {
            while (true) {
                if (offer(task, 500, TimeUnit.MILLISECONDS)) {
                    log.warn("添加任务，循环 offer 直至成功入列，计数 {}", realPutDelayCount.intValue());
                    break;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 添加私信队列任务
     * @param task 实现 Runnable 接口的任务
     */
    public static void putDead(Runnable task) {
        try {
            if (LIMITER_PUT.tryAcquire()) {
                DEAD_QUEUE.put(task);
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 等待并重新添加任务
     * @param task 实现 Runnable 接口的任务
     * @param timeout 放弃前等待多长时间
     * @param unit 确定如何解释timeout参数
     * @return true 如果成功，或者 false 指定的等待时间已过，则有可用空间
     */
    public static boolean offer(Runnable task, long timeout, TimeUnit unit) {
        try {
            if (LIMITER_PUT.tryAcquire() && QUEUE.offer(task, timeout, unit)) {
                realPutDelayCount.incrementAndGet();
                return true;
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return false;
    }

    /**
     * 重试3次，每隔30秒重试一次
     * @param task 实现 Runnable 接口的任务
     */
    public static void retry(Runnable task) {
        log.warn("任务[{}]正在重试！", task);
        boolean result = false;
        for (int i = 0; i < RETRY_TIMES; i++) {
            result = offer(task, 30, TimeUnit.SECONDS);
            if (result) {
                break;
            }
        }

        if (!result) {
            // 放入私信队列
            putDead(task);
            // 添加日志（加入数据库）
            log.info("任务[{}]已重试{}次", task, RETRY_TIMES);
        }
    }

    /**
     * 开启一个工人线程 不停拉取消费队列任务
     * @param i 序号
     */
    public static void submit(int i) {
        SERVICE.submit(() -> {
            Runnable task;
//          while (true) {// 这里存在死循环，即使手动shutdown 进程也不会终止
            while (!Thread.currentThread().isInterrupted()) {
//              task = poll();
                task = pollNew();
                if (task != null) {
                    // run 方法必须捕获异常
                    try {
                        task.run();
                    } catch (Exception e) {
//                      retry(task);
                    } finally {
                        realPollCount.incrementAndGet();
                    }
                }
            }
        });
        log.info("工人[{}]已上线，正在加班加点工作。。", i);
    }

    /**
     * 关闭工人线程池
     */
    public static void shutDown() {
        log.info("正在关闭工人线程池资源。。");
        try {
            SERVICE.shutdown();
            if (!SERVICE.awaitTermination(5, TimeUnit.SECONDS)) {
                SERVICE.shutdownNow();
                int taskCount = count();
                int taskRealPutCount = realPutCount.intValue();
                int taskRealPollCount = realPollCount.intValue();
                log.info("已关闭工人线程池资源！共计接收任务{}个，实际接收任务{}个，实际消费任务{}个", taskCount, taskRealPutCount, taskRealPollCount);
                // 清空队列
                int size = QUEUE.size();
                QUEUE.clear();
                log.info("已关闭任务队列资源！当前队列{}个", size);
                int deadSize = DEAD_QUEUE.size();
                DEAD_QUEUE.clear();
                log.info("已关闭任务私信队列资源！当前死信队列{}个", deadSize);
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
